package com.qqt.csr.im.service.mongo;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.mongodb.client.result.UpdateResult;
import com.qqt.csr.common.sensitiveword.SensitiveWordMsg;
import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.common.utils.SnowflakeUtil;
import com.qqt.csr.im.assembler.MessageAssembler;
import com.qqt.csr.im.assembler.MessageMapping;
import com.qqt.csr.im.dto.MsgAccountDTO;
import com.qqt.csr.im.dto.SendSystemMessageDTO;
import com.qqt.csr.im.dto.SensitiveWordResultDTO;
import com.qqt.csr.im.entity.Account;
import com.qqt.csr.im.entity.Connection;
import com.qqt.csr.im.entity.RoomMember;
import com.qqt.csr.im.entity.mq.MqMessage;
import com.qqt.csr.im.entity.msg.*;
import com.qqt.csr.im.enums.MsgFlowTypeEnum;
import com.qqt.csr.im.enums.MsgTypeEnum;
import com.qqt.csr.im.enums.ReadFlagEnum;
import com.qqt.csr.common.sensitiveword.SensitiveWordShowTypeEnum;
import com.qqt.csr.im.mapper.mongo.MessageMapper;
import com.qqt.csr.im.service.AccountService;
import com.qqt.csr.im.service.ConnectionService;
import com.qqt.csr.im.service.EventMsgProducerService;
import com.qqt.csr.im.service.sensitiveword.SensitiveWordService;
import com.qqt.csr.im.socketio.constants.EventsConstants;
import com.qqt.csr.im.utils.DeviceCacheKeyUtil;
import com.qqt.csr.im.vo.req.ConversationReadReqVO;
import com.qqt.csr.im.vo.req.OfflineMessageReqVO;
import com.qqt.csr.im.vo.resp.MessageRespVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.*;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @Author: yang hao
 * @Date: 2024/10/15 9:44
 * @Description: 消息 业务层
 **/
@Slf4j
@Service
public class MessageService {
    @Autowired
    private MessageMapper messageMapper;
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private ConversationService conversationService;
    @Autowired
    private AccountService accountService;
    @Autowired
    private ConnectionService connectionService;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private EventMsgProducerService producerService;
    @Autowired
    private SensitiveWordService sensitiveWordService;

    /**
     * 保存副本消息
     *
     * @param senderAccountId
     * @param originalMsg
     * @param roomMemberList
     * @return
     */
    public List<Message> saveDuplicateMsg(Long senderAccountId, Message originalMsg, List<RoomMember> roomMemberList) {
        // 敏感词检测
        SensitiveWordResultDTO swResultDTO = null;
        if (originalMsg.getContent() instanceof TextContent textContent && MsgTypeEnum.TextElem.equals(originalMsg.getType())) {
            Account sender = accountService.getMapByIds(Lists.newArrayList(senderAccountId)).get(senderAccountId);
            swResultDTO = sensitiveWordService.filter(originalMsg.getTenantId(), textContent.getText(), sender);
        }

        // 创建副本消息，包括推送人，保存MongoDB; 同一条消息不同接收人的副本消息的序号都是一样
        List<Message> saveMsgList = new ArrayList<>();
        for (RoomMember roomMember : roomMemberList) {
            Message copyMsg = MessageMapping.INSTANCE.copy(originalMsg);
            copyMsg.setRecipient(MsgAccountDTO.builder().id(roomMember.getAccountId().toString()).build());
            if (copyMsg.getSender().getId().equals(copyMsg.getRecipient().getId())) {
                // 自身消息设置为已读
                copyMsg.setReadFlag(ReadFlagEnum.READ);
                copyMsg.setFlow(MsgFlowTypeEnum.OUT);
            } else {
                copyMsg.setReadFlag(ReadFlagEnum.UNREAD);
                copyMsg.setFlow(MsgFlowTypeEnum.IN);
            }

            if (swResultDTO == null) {
                // 无命中敏感词
                copyMsg.setSensitiveWords(Lists.newArrayList());
                saveMsgList.add(copyMsg);
            } else {
                // 记录敏感词检测结果
                List<SensitiveWordMsg> sensitiveWordMsgList = new ArrayList<>();
                Integer showType = swResultDTO.getShowType();
                if (SensitiveWordShowTypeEnum.ASTERISKS_DISPLAYED.getCode().equals(showType)) {
                    // 双方星号加密
                    sensitiveWordMsgList = swResultDTO.getWords().stream().map(wordResult -> SensitiveWordMsg.of(showType, wordResult.getStartIndex(), wordResult.getEndIndex())).collect(Collectors.toList());
                } else if (SensitiveWordShowTypeEnum.HIGHLIGHTED_DISPLAYED.getCode().equals(showType)) {
                    if (senderAccountId.equals(roomMember.getAccountId())) {
                        // 只有发送方高亮显示，接收方不需要接收此条消息
                        sensitiveWordMsgList = swResultDTO.getWords().stream().map(wordResult -> SensitiveWordMsg.of(showType, wordResult.getStartIndex(), wordResult.getEndIndex())).collect(Collectors.toList());
                    }
                }
                if (CollectionUtils.isNotEmpty(sensitiveWordMsgList)) {
                    copyMsg.setSensitiveWords(sensitiveWordMsgList);
                    saveMsgList.add(copyMsg);
                }
            }
        }

        // 保存副本消息
        messageMapper.saveAll(saveMsgList);
        return saveMsgList;
    }

    /**
     * 拉取离线消息列表
     *
     * @param dto {@link OfflineMessageReqVO} 离线消息dto
     * @return {@link List <MessageVO>} 离线消息列表
     */
    public List<MessageRespVO> pullOfflineMessage(OfflineMessageReqVO dto) {
        Criteria criteria = Criteria.where("conversationId").is(dto.getConversationId());
        if (dto.getAccountId() != null) {
            criteria.and("recipient._id").is(dto.getAccountId());
        }
        MatchOperation matchOperation = Aggregation.match(criteria);
        SortOperation sortOperation = Aggregation.sort(Sort.Direction.ASC, "seq");
        Aggregation aggregation = Aggregation.newAggregation(matchOperation, sortOperation);
        AggregationResults<Message> result = mongoTemplate.aggregate(aggregation, "im_message", Message.class);

        List<Message> messageList = result.getMappedResults();
        for (Message message : messageList) {
            // 敏感词替换，推送到接收方展示
            if (CollectionUtils.isNotEmpty(message.getSensitiveWords())
                    && MsgTypeEnum.TextElem.equals(message.getType())) {
                TextContent swTextContent = JsonUtil.toObject(JsonUtil.toJson(message.getContent()), TextContent.class);
                Optional.ofNullable(swTextContent).ifPresent(content -> {
                    swTextContent.setText(sensitiveWordService.replace(swTextContent.getText(), message.getSensitiveWords()));
                    message.setContent(swTextContent);
                });
            }
        }
        return MessageAssembler.convert(messageList);
    }

    /**
     * 未读数
     *
     * @param accountId
     * @return {@link Long} 未读数
     */
    public long getNoReadCount(String conversationId, String accountId) {
        Criteria criteria = Criteria.where("conversationId").is(conversationId)
                .and("readFlag").is(ReadFlagEnum.UNREAD.name())
                .and("recipient._id").is(accountId);
        return mongoTemplate.count(Query.query(criteria), Message.class);
    }

    /**
     * 设置已读
     *
     * @param dto {@link ConversationReadReqVO} 已读参数
     * @return {@link Boolean} 是否成功
     */
    public boolean read(ConversationReadReqVO dto) {
        Criteria criteria = Criteria.where("conversationId").is(dto.getConversionId());
        if (StringUtils.isNotBlank(dto.getSeq())) {
            criteria.and("seq").lte(dto.getSeq());
        }
        criteria.and("recipient._id").is(dto.getAccountId().toString())
                .and("readFlag").is(ReadFlagEnum.UNREAD.name());
        UpdateResult updateResult = mongoTemplate.updateMulti(Query.query(criteria), Update.update("readFlag", ReadFlagEnum.READ.name()), Message.class);
        return updateResult.getModifiedCount() > 0;
    }

    /**
     * 发送系统消息
     *
     * @param dto {@link SendSystemMessageDTO} 发送系统消息
     * @return {@link Boolean} 是否发送成功
     */
    public boolean sendSystemMsg(SendSystemMessageDTO dto, Account account) {
        Map<String, Account> accountMap = Maps.newHashMap();
        if (account != null && account.getAccountUserId() != null) {
            accountMap.put(account.getAccountUserId(), account);
        }
        sendSystemMsg(Lists.newArrayList(dto), accountMap);
        return true;
    }

    /**
     * 批量推送系统消息
     *
     * @param dtoList
     * @param accountMap
     */
    public void sendSystemMsg(List<SendSystemMessageDTO> dtoList, Map<String, Account> accountMap) {
        if (CollectionUtils.isEmpty(dtoList)) {
            return;
        }

        if (MapUtils.isEmpty(accountMap)) {
            List<String> userIdFlagList = dtoList.stream().map(SendSystemMessageDTO::getUserId)
                    .filter(StringUtils::isNotBlank).distinct().collect(Collectors.toList());
            accountMap = accountService.getMapByUserIds(userIdFlagList);
        }

        List<Long> accountIdList = accountMap.values().stream().map(Account::getId).collect(Collectors.toList());
        Map<Long, List<Connection>> connectionMap = connectionService.getOnlineByAccountId(accountIdList);

        List<Message> messageList = new ArrayList<>();
        List<MqMessage> mqMessageList = new ArrayList<>();

        for (SendSystemMessageDTO sendSystemMessageDTO : dtoList) {
            Account account = accountMap.get(sendSystemMessageDTO.getUserId());
            if (account == null) {
                log.info("无对应的im账号，systemMsg:{}", sendSystemMessageDTO);
                continue;
            }
            Message message = MessageAssembler.convert(sendSystemMessageDTO, account);
            messageList.add(message);

            List<Connection> connections = connectionMap.get(account.getId());
            if (CollectionUtils.isEmpty(connections)) {
                log.info("im账号无连接信息，systemMsg:{}，accountId:{}", sendSystemMessageDTO, account.getId());
                continue;
            }
            for (Connection connection : connections) {
                String deviceKey = DeviceCacheKeyUtil.getKey(connection);
                MqMessage mqData = MqMessage.builder().serviceId(connection.getServiceId()).deviceKey(deviceKey).uniqueKey(SnowflakeUtil.nextIdStr()).message(message).build();
                mqMessageList.add(mqData);
            }
        }

        if (CollectionUtils.isNotEmpty(messageList)) {
            messageMapper.saveAll(messageList);
        }

        if (CollectionUtils.isNotEmpty(mqMessageList)) {
            mqMessageList.forEach(mqMessage -> producerService.sendEventMsg(EventsConstants.S2S_MESSAGE_EVENT, mqMessage));
        }
    }

    /**
     * 获取每个会话中消息发送方最后一条消息
     *
     * @param conversationIds
     * @return key: senderConnectId_conversationId
     */
    public Map<String, Message> getEverySenderLastMsgBy(List<String> conversationIds) {
        MatchOperation matchOperation = Aggregation.match(Criteria.where("conversationId").in(conversationIds));
        SortOperation sortOperation = Aggregation.sort(Sort.Direction.DESC, "seq");
        GroupOperation groupOperation = Aggregation.group("sender._id", "conversationId")
                .first("$$ROOT").as("latestMessage");
        ReplaceRootOperation replaceRootOperation = Aggregation.replaceRoot("latestMessage");
        ProjectionOperation projectionOperation = Aggregation.project("_id", "seq", "sender", "conversationId", "createTime");
        Aggregation aggregation = Aggregation.newAggregation(
                matchOperation,
                sortOperation,
                groupOperation,
                replaceRootOperation,
                projectionOperation
        );

        AggregationResults<Message> result = mongoTemplate.aggregate(aggregation, "im_message", Message.class);
        return result.getMappedResults().stream()
                .filter(message -> message.getSender() != null && message.getSender().getId() != null)
                .collect(Collectors.toMap(k -> k.getSender().getId() + "_" + k.getConversationId(), v -> v, (o, n) -> o));
    }

    /**
     * 转换消息内容类型
     *
     * @param originalMsg
     * @return
     */
    private Content convertMsgContent(Message originalMsg) {
        Content content;
        if (MsgTypeEnum.SystemElem.equals(originalMsg.getType()) || MsgTypeEnum.TextElem.equals(originalMsg.getType())) {
            content = JsonUtil.toObject(JsonUtil.toJson(originalMsg.getContent()), TextContent.class);
        } else if (MsgTypeEnum.ImageElem.equals(originalMsg.getType())) {
            content = JsonUtil.toObject(JsonUtil.toJson(originalMsg.getContent()), ImageContent.class);
        } else {
            content = JsonUtil.toObject(JsonUtil.toJson(originalMsg.getContent()), FileContent.class);
        }
        return content;
    }

}
